package com.surfilter.msmp.web.system.scheduled;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.surfilter.msmp.common.util.HttpUtil;
import com.surfilter.msmp.web.es.EsIndexName;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.ElasticsearchStatusException;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.ClearScrollRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchScrollRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.Scroll;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.FieldSortBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 描述 ：封堵日志全量验证任务
 * 作者 ：WYH
 * 时间 ：2019/11/26 16:36
 **/
@Slf4j
@Component
@EnableScheduling
public class BlockLogCheck {

    @Autowired
    private RestHighLevelClient highLevelClient;

    private AtomicInteger atomicInteger = new AtomicInteger(0);

    /**
     * 每天凌晨2点执行一次
     * es限制每次查询最多10000条数据 要爬取大量数据 需通过游标查询方式
     * 游标不适用于实时搜索
     */
    @Scheduled(cron = "0 0 2 * * ?")
    public void job() {
        log.info("近24小时内封堵日志验证任务执行开始...");
        try {
            int count = 0;
            final Scroll scroll = new Scroll(TimeValue.timeValueMinutes(1L));
            //SearchRequest searchRequest = new SearchRequest("block_log_test");//测试索引
            SearchRequest searchRequest = new SearchRequest(EsIndexName.MSMP_BLOCK_LOG.getIndexName());
            searchRequest.scroll(scroll);

            //查询24小时内的
            BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
            DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
            LocalDateTime dateTime = LocalDateTime.now().minusDays(1);
            String timeString = dateTime.format(formatter);
            boolQueryBuilder.must(QueryBuilders.rangeQuery("verifyTime.keyword").from(timeString));

            //每个线程处理100个 并行度50
            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
            searchSourceBuilder.query(boolQueryBuilder).size(100)
                    .sort(new FieldSortBuilder("verifyTime.keyword").order(SortOrder.DESC));
            searchRequest.source(searchSourceBuilder);

            SearchResponse searchResponse;
            try {
                searchResponse = highLevelClient.search(searchRequest, RequestOptions.DEFAULT);
            } catch (ElasticsearchStatusException e) {
                log.warn("查询出错：{}", e.getMessage());
                return;
            }
            String scrollId = searchResponse.getScrollId();
            SearchHit[] searchHits = searchResponse.getHits().getHits();
            ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("LogCheck-task-%d").build();
            //todo 此处因最大任务量不明确 线程池参数暂未确定 目前假定每天最多1.5w条
            ThreadPoolExecutor executor = new ThreadPoolExecutor(50, 50, 1, TimeUnit.MINUTES, new LinkedBlockingQueue<>(100), threadFactory, new ThreadPoolExecutor.CallerRunsPolicy());

            while (searchHits != null && searchHits.length > 0) {
                while (executor.getQueue().remainingCapacity() <= 0) {
                    //队列满了说明线程池满了 等候10秒再试
                    log.debug("线程池已满 等候10s");
                    Thread.sleep(10000);
                }
                count += searchHits.length;
                executor.execute(new BatchCheckTask(searchHits));

                SearchScrollRequest scrollRequest = new SearchScrollRequest(scrollId);
                scrollRequest.scroll(scroll);
                searchResponse = highLevelClient.scroll(scrollRequest, RequestOptions.DEFAULT);
                scrollId = searchResponse.getScrollId();
                searchHits = searchResponse.getHits().getHits();
            }
            log.debug("目前累计下发日志数: {}", count);

            //执行完成后清理资源
            ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
            clearScrollRequest.addScrollId(scrollId);
            highLevelClient.clearScroll(clearScrollRequest, RequestOptions.DEFAULT);
            executor.shutdown();
            while (!executor.isTerminated()) {
                log.debug("目前执行完任务数：" + atomicInteger.get() + " 继续等待线程池执行10秒");
                Thread.sleep(10000);
            }
            log.info("封堵日志验证任务执行结束,本次共验证数量：{}", atomicInteger.get());
            atomicInteger.set(0);
        } catch (Exception e) {
            log.error("封堵日志验证任务出错：{}", e.getMessage());
        }
    }


    class BatchCheckTask implements Runnable {
        private SearchHit[] hits;

        BatchCheckTask(SearchHit[] hits) {
            this.hits = hits;
        }

        @Override
        public void run() {
            if (hits.length == 0) {
                return;
            }
            try {
                //使用bulk请求 批量处理 减小访问次数
                BulkRequest bulkRequest = new BulkRequest();
                for (SearchHit hit : hits) {
                    String string = hit.getSourceAsString();
                    JSONObject blockLog = JSON.parseObject(string);
                    String url = blockLog.getString("url");
                    if (StringUtils.isBlank(url)) {
                        continue;
                    }
                    IndexRequest request = new IndexRequest(EsIndexName.MSMP_CHECK_LOG.getIndexName(), "doc");
                    DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
                    LocalDateTime dateTime = LocalDateTime.now();
                    String timeString = dateTime.format(formatter);
                    Map<String, Object> source = new HashMap<>(8);
                    //先将原始url加入map
                    source.put("url", url);
                    if (!url.contains("http")) {
                        url = "http://" + url;
                    }
                    int effect = HttpUtil.isEffect(url, 5000);
                    source.put("checkResult", (effect != 1));
                    source.put("checkTime", timeString);
                    request.source(source);
                    bulkRequest.add(request);
                    atomicInteger.incrementAndGet();
                }
                highLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
            } catch (Exception e) {
                log.error(Thread.currentThread().getName() + "执行出错：{}", e.getMessage());
            }
        }
    }

}
